/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.wx.mq.redisThread;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import redis.clients.jedis.Jedis;
import cn.ac.iie.di.wx.mq.common.RuntimeEnv;
import cn.ac.iie.di.wx.mq.utils.JedisConnectionHandler;
import redis.clients.jedis.Pipeline;

/**
 *
 * @author Administrator
 */
public class JedisThread implements Runnable {

    public LinkedBlockingQueue<HashMap<String, String>> redisQueue;
    public LinkedBlockingQueue<HashMap<String, String>> urgent_redisQueue;
    private static final Logger logger = Logger.getLogger(JedisThread.class.getName());

    public JedisThread(LinkedBlockingQueue<HashMap<String, String>> redisQueue) {
        this.redisQueue = redisQueue;
    }

    public JedisThread(LinkedBlockingQueue<HashMap<String, String>> redisQueue, LinkedBlockingQueue<HashMap<String, String>> urgent_redisQueue) {
        this.redisQueue = redisQueue;
        this.urgent_redisQueue = urgent_redisQueue;
    }

    @Override
    public void run() {
        Jedis jedis = null;
        JedisConnectionHandler jedisConn = null;
        try {

//            String[] ipAndPort = ((String) RuntimeEnv.getParam(RuntimeEnv.redis_hosts)).split(":");
            jedisConn = new JedisConnectionHandler((String) RuntimeEnv.getParam(RuntimeEnv.redis_hosts),
                    (String) RuntimeEnv.getParam(RuntimeEnv.redis_masterName),
                    Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.redis_dbName)));

//            logger.info(jedis);
        } catch (Exception e) {
            logger.fatal(e.getMessage(), e);
            System.exit(-1);
        }
        int pipMaxNum = Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.pipMaxNum));
        logger.info("set redis pipMaxNum = " + pipMaxNum);
        while (true) {
            try {
                HashMap<String, String> redisMap = new HashMap<>();
                for (int i = 0; i < pipMaxNum; i++) {
                    //紧急队列优先
                    HashMap<String, String> hashMap = null;

                    if (urgent_redisQueue.size() > 0) {
                        hashMap = urgent_redisQueue.poll(50, TimeUnit.MILLISECONDS);
                    } else {
                        hashMap = redisQueue.poll(50, TimeUnit.MILLISECONDS);
                    }
                    if (hashMap != null) {
                        for (Map.Entry<String, String> entrySet : hashMap.entrySet()) {
                            redisMap.put(entrySet.getKey(), entrySet.getValue());
                        }
                    }
                }
                logger.info("put into redisMap size = " + redisMap.size());
                //将map的值进行批量入redis
                //失败的重试
                while (true) {
                    try {
                        jedis = jedisConn.getJedis();
                        Pipeline pip = jedis.pipelined();
                        long redisStarttime = System.currentTimeMillis();
                        for (Map.Entry<String, String> redisEntrySet : redisMap.entrySet()) {
                            pip.setex(redisEntrySet.getKey(), 60 * 60 * 24 * 7, redisEntrySet.getValue());
                            logger.info("set into redis:" + redisEntrySet.getKey() + ",value = " + redisEntrySet.getValue());
                        }
                        pip.sync();
                        logger.info("pip setex use time = " + (System.currentTimeMillis() - redisStarttime) + " ms");
                        try {
                            Thread.sleep(1);
                        } catch (Exception e) {
                            logger.error(e.getMessage(), e);
                        }
                        break;
                    } catch (Exception ex) {
                        logger.error("pip setex error ,will retry now" + ex.getMessage(), ex);
                        try {
                            Thread.sleep(100);
                        } catch (Exception e) {
                            logger.error(e.getMessage(), e);
                        }
                        //重新获取连接
                        if (jedis != null) {
                            jedisConn.returnResource(jedis);
                        }
                        jedis = jedisConn.getJedis();
                    }

                }
            } catch (InterruptedException ex) {
                logger.error(ex.getMessage(), ex);
            } finally {
                if (jedis != null) {
                    jedisConn.returnResource(jedis);
                }
            }
        }
    }

}
